package sparkcore.day3.lesson01;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * Created by Administrator on 2018/4/24.
 */
public class ActionOperator {
    public static SparkConf conf = new SparkConf().setMaster("local").setAppName("test");
    public static JavaSparkContext sc = new JavaSparkContext(conf);
    /***
     * max,min    => .max  .min
     * TopN
     * 二次排序
     *
     *
     */
    /**
     * 取出单词次数出现最多的前两个单词
     */
    public static void topN(){
        final JavaRDD<String> rdd = sc.parallelize(
                Arrays.asList("you,jump", "you,jump", "i,you"));
        final List<Tuple2<Integer, String>> result = rdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(",")).iterator();
            }
        }).mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<>(word, 1);
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }).mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {
                return new Tuple2<>(tuple._2, tuple._1);
            }
        }).sortByKey(false).take(2);

       for(Tuple2<Integer, String> tuple:result){
           System.out.println(tuple._2 + "  "+ tuple._1);
       }


    }

    /**
     *
     * @param args
     */

    public static void main(String[] args) {
        topN();

    }
}
